package com.lry.read;

import com.lry.Node;
import com.lry.NodeValue;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.LongAdder;

@Slf4j
public abstract class AbstractReadConsistency implements ReadConsistency {

    protected List<CompletableFuture<NodeValue>> getAsync(String key, List<Node> nodes) {
        List<CompletableFuture<NodeValue>> futures = new ArrayList<>();
        for (Node node : nodes) {
            CompletableFuture future = CompletableFuture.supplyAsync(()-> node.get(key));
            futures.add(future);
        }
        return futures;
    }

//    protected NodeValue getReturn(List<CompletableFuture<NodeValue>> futures, int requireSuccessNum){
//        LongAdder successNum = new LongAdder();
//
//        Map<Integer, NodeValue> map = new TreeMap<>(Comparator.reverseOrder());
//        CountDownLatch cdl = new CountDownLatch(1);
//
//        for (CompletableFuture<NodeValue> future : futures) {
//            //多线程防止第一个由于网络问题卡住，获取结果，执行自增
//              CompletableFuture.runAsync(()->{
//                try {
//                    NodeValue nodeValue = future.get();
//                    if(nodeValue!=null){
////                        Thread.sleep(10); //在readOne级别下，当node1 error的时候，put失败，当get的时候先访问node1，successNum自增，测试node2 休眠10ms，会看到从node1拿不到数据
//                        map.put(nodeValue.getVersion(), nodeValue);
//                    }
//
//                    successNum.increment();
//                    if(successNum.intValue()>=requireSuccessNum){
//                        cdl.countDown();
//                    }
//
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                } catch (ExecutionException e) {
//                    e.printStackTrace();
//                }
//            });
//        }
//
//        try {
//            cdl.await();
//        } catch (InterruptedException e) {
//            log.info("读数据超时");
//            e.printStackTrace();
//            return null;
//        }
//
//
//        //拿版本号最大的
//        Iterator<Map.Entry<Integer,NodeValue>> it = map.entrySet().iterator();
//        log.info("map={}",map);
//        if(!it.hasNext()){
//            return null;
//        }
//        return  it.next().getValue();
//
//    }

    protected NodeValue getReturn(List<CompletableFuture<NodeValue>> futures, int requireSuccessNum){
        LongAdder successNum = new LongAdder();

        Map<Integer, NodeValue> map = new TreeMap<>(Comparator.reverseOrder());

        for (CompletableFuture<NodeValue> future : futures) {
            //多线程防止第一个由于网络问题卡住，获取结果，执行自增
            CompletableFuture.runAsync(()->{
                try {
                    NodeValue nodeValue = future.get();
                    if(nodeValue!=null){
//                        Thread.sleep(10); //在readOne级别下，当node1 error的时候，put失败，当get的时候先访问node1，successNum自增，测试node2 休眠10ms，会看到从node1拿不到数据
                        map.put(nodeValue.getVersion(), nodeValue);
                    }

                    successNum.increment();

                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            });
        }

        while(successNum.intValue()<requireSuccessNum){
            try {
                Thread.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }


        //拿版本号最大的
        Iterator<Map.Entry<Integer,NodeValue>> it = map.entrySet().iterator();
        log.info("map={}",map);
        if(!it.hasNext()){
            return null;
        }
        return  it.next().getValue();

    }


}
